package com.boguan.web.controller.pulsar;

import com.alibaba.fastjson.JSONObject;
import com.boguan.common.utils.StringUtils;
import com.boguan.common.utils.spring.SpringUtils;
import com.boguan.device.domain.*;
import com.boguan.device.mapper.DeviceMapper;
import com.boguan.device.service.IDeviceAlarmService;
import com.boguan.device.service.IDeviceService;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.shade.com.google.common.primitives.Chars;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;

//@Component
public class PulsarConsumer {

    private static final Logger log = LoggerFactory.getLogger(PulsarConsumer.class);

    static String appId = "2510";
    static String appSecret = "1854c2255e2f4b3545c96c9c71b2cd03f5f04512ddcdbc601d5245d0355adce4";
    static String serviceUrl = "pulsar://pulsar.weiyuzn.com:6650";


    private PulsarClient client = null;
    private Consumer consumer = null;

    /**
     * 使用@PostConstruct注解用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化
     */
    @PostConstruct
    public void initPulsar() throws Exception {
        try {
            //构造Pulsar client
            this.client = PulsarClient.builder()
                    .serviceUrl(serviceUrl)
                    .authentication(new WeiYuAuthentication(appId, appSecret))
                    .build();

            //创建consumer
            this.consumer = client.newConsumer()
                    .topic("persistent://" + appId + "/out/deviceState")
                    .subscriptionName(appId)
                    .subscriptionType(SubscriptionType.Key_Shared)
                    .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(3).build())
                    .subscribe();

            // 开始消费
            new Thread(() -> {
                try {
                    while (true) {
                        start();
                    }
                } catch (Exception e) {
                    log.error("消费Pulsar数据异常，停止Pulsar连接：", e);
                    close();
                }
            }).start();

        } catch (Exception e) {
            log.error("Pulsar初始化异常：", e);
            throw e;
        }
    }

    public void start() throws Exception {
        //消费消息
        while (true) {
            Message message = consumer.receive();
//            String[] keyArr = message.getKey().split("_");
            String jsons = new String(message.getData());
            System.out.println(jsons);
            if (StringUtils.isNotEmpty(jsons)) {
                try {
                    wsSend(jsons);
                } catch (Exception e) {
                    log.error("消费Pulsar数据异常，key【{}】，json【{}】：", message.getKey(), jsons, e);
                }
            }
            consumer.acknowledge(message);
        }
    }


    /**
     * 线程池异步处理Pulsar推送的数据
     */
    @Async("threadPoolTaskExecutor")
    @Transactional
    public void wsSend(String message) {
        JSONObject jsonObject = JSONObject.parseObject(message);
        if (StringUtils.isNotNull(jsonObject)) {
            String bizCode = jsonObject.getString("bizCode");
            JSONObject bizData = jsonObject.getJSONObject("bizData");
            if (StringUtils.isNotNull(bizData)) {
                String devno = bizData.getString("devno");
                IDeviceService deviceService = SpringUtils.getBean(IDeviceService.class);
                Device device = deviceService.selectDeviceByDeviceNum(devno, 0);
//                long time = bizData.getLong("time") * 1000L;
                Date dataTime = new Date();
                if (StringUtils.isNotNull(device)) {
                    IDeviceAlarmService deviceAlarmService = SpringUtils.getBean(IDeviceAlarmService.class);
                    switch (bizCode) {
                        case "status":
                            List<DeviceParam> deviceParams = device.getDeviceParams();
                            Integer state = bizData.getInteger("state");
                            device.setState(state);
                            device.setSignalQuality(bizData.getString("CSQ"));
                            device.setOnlineStatus(1);
                            for (DeviceParam deviceParam : deviceParams) {
                                Double value = bizData.getDouble(deviceParam.getIdentifier());

                                if (value != null) {
                                    double v = value * deviceParam.getScale();
                                    String s = BigDecimal.valueOf(v).setScale(2, RoundingMode.FLOOR).toString();
                                    if (deviceParam.getId() != null) {
                                        //删除原来的key
                                        deviceParam.setValue(s);
                                        deviceParam.setTypeId(device.getTypeId());
                                        deviceParam.setDeviceId(device.getDeviceId());
                                        deviceService.updateDeviceParam(deviceParam);
                                    } else {
                                        deviceParam.setValue(s);
                                        deviceParam.setTypeId(device.getTypeId());
                                        deviceParam.setDeviceId(device.getDeviceId());
                                        deviceService.insertSelective(deviceParam);
                                    }
                                }
                                if (StringUtils.isNull(deviceParam.getValue())) {
                                    deviceParam.setValue("0");
                                }
                                DevicePastRecords pastRecords = new DevicePastRecords();
                                pastRecords.setDeviceId(device.getDeviceId());
                                pastRecords.setIdentifier(deviceParam.getIdentifier());
                                pastRecords.setParamId(deviceParam.getParamId());
                                pastRecords.setParamName(deviceParam.getParamName());
                                pastRecords.setParamType(deviceParam.getParamType().toString());
                                pastRecords.setTime(dataTime);
                                pastRecords.setValue(deviceParam.getValue());
                                deviceService.insertSelectivePastRecords(pastRecords);
                            }
                            /**
                             * 报警信息
                             */
                            DeviceParamVo deviceParamVo = deviceService.selectDeviceParam(device.getDeviceId(), 2);
                            List<DeviceParam> alarmParams = deviceParamVo.getDeviceParams();
                            if (StringUtils.isNotNull(alarmParams) && alarmParams.size() > 0) {
                                Long fault = bizData.getLong("fault");
                                Long alarm = bizData.getLong("alarm");
                                for (DeviceParam deviceParam : alarmParams) {
                                    String identifier = deviceParam.getIdentifier();
                                    int index = Integer.parseInt(identifier.substring(identifier.lastIndexOf("_") + 1));
                                    if (identifier.startsWith("alarm")) {
                                        /**
                                         * 报警
                                         */
                                        if (StringUtils.isNotNull(alarm) && alarm > 0) {
                                            long i = alarm >> index & 1;
                                            if (i == 1) {
                                                /**
                                                 * 报警
                                                 */
                                                DeviceAlarmLog deviceAlarmLogs = new DeviceAlarmLog();
                                                deviceAlarmLogs.setDeviceId(device.getDeviceId());
                                                deviceAlarmLogs.setAlarmTime(dataTime);
                                                deviceAlarmLogs.setAlarmLevel(3);
                                                deviceAlarmLogs.setErrorState(1);
                                                deviceAlarmLogs.setAlarmCause(deviceParam.getParamName());
                                                deviceAlarmLogs.setIdentifier(deviceParam.getIdentifier());
                                                deviceAlarmLogs.setParamName(deviceParam.getParamName());
                                                deviceAlarmLogs.setValue("");
                                                deviceAlarmLogs.setStatus(1);
                                                deviceAlarmLogs.setAlarmLocation(device.getLocaltion());
                                                deviceAlarmLogs.setCreateTime(dataTime);
                                                deviceAlarmService.insertSelective(deviceAlarmLogs);
                                            }
                                        }
                                    } else {
                                        /**
                                         * 故障
                                         */
                                        if (StringUtils.isNotNull(fault) && fault > 0) {
                                            long i = fault >> index & 1;
                                            if (i == 1) {
                                                /**
                                                 * 故障
                                                 */
                                                DeviceAlarmLog deviceAlarmLogs = new DeviceAlarmLog();
                                                deviceAlarmLogs.setDeviceId(device.getDeviceId());
                                                deviceAlarmLogs.setAlarmTime(dataTime);
                                                deviceAlarmLogs.setAlarmLevel(4);
                                                deviceAlarmLogs.setErrorState(1);
                                                deviceAlarmLogs.setAlarmCause(deviceParam.getParamName());
                                                deviceAlarmLogs.setIdentifier(deviceParam.getIdentifier());
                                                deviceAlarmLogs.setParamName(deviceParam.getParamName());
                                                deviceAlarmLogs.setValue("");
                                                deviceAlarmLogs.setStatus(1);
                                                deviceAlarmLogs.setAlarmLocation(device.getLocaltion());
                                                deviceAlarmLogs.setCreateTime(dataTime);
                                                deviceAlarmService.insertSelective(deviceAlarmLogs);
                                            }
                                        }
                                    }
                                }
                            }
                            break;
                        case "online":
                            /**
                             * 上线
                             */
                            device.setOnlineStatus(1);
                            break;
                        case "offline":
                            /**
                             * 离线
                             */
                            device.setOnlineStatus(2);
                            /**
                             * 设备离线
                             */
                            DeviceAlarmLog deviceAlarmLogs = new DeviceAlarmLog();
                            deviceAlarmLogs.setDeviceId(device.getDeviceId());
                            deviceAlarmLogs.setAlarmTime(dataTime);
                            deviceAlarmLogs.setAlarmLevel(4);//故障
                            deviceAlarmLogs.setErrorState(1);//报警状态
                            deviceAlarmLogs.setAlarmCause("设备离线");
                            deviceAlarmLogs.setIdentifier(device.getDeviceNum() + "offline");
                            deviceAlarmLogs.setParamName("设备离线");
                            deviceAlarmLogs.setValue("");
                            deviceAlarmLogs.setStatus(1);//处理状态
                            deviceAlarmLogs.setAlarmLocation(device.getLocaltion());
                            deviceAlarmLogs.setCreateTime(dataTime);
                            deviceAlarmService.insertSelective(deviceAlarmLogs);
                            break;
                        default:
                            break;

                    }
                    deviceService.updateDevice(device);
                }
            }
        }

    }


    public void close() {
        try {
            consumer.close();
        } catch (PulsarClientException e) {
            log.error("关闭Pulsar消费者失败：", e);
        }
        try {
            client.close();
        } catch (PulsarClientException e) {
            log.error("关闭Pulsar连接失败：", e);
        }
    }

    public static void main(String[] args) {
        //数据包含温度故障与外部故障
        long fault = 255;
        HashMap<Integer, String> qf1FaultMap = new HashMap<Integer, String>();
        qf1FaultMap.put(0, "过载故障");
        qf1FaultMap.put(1, "瞬时故障");
        qf1FaultMap.put(6, "温度故障");
        qf1FaultMap.put(7, "外部故障");

        List<Character> bits = Chars.asList(Long.toBinaryString(fault).toCharArray());
//        System.out.println(bits.size());
        Collections.reverse(bits);

        qf1FaultMap.forEach((k, v) -> {
            long i = fault >> k & 1;
            System.out.println(i);
            if (i == 1) {
                System.out.println(v);
            }


        });

    }


}
